package com.audaque.springboot.foshanupload.web.amqpdemo.controller;

import cn.hutool.json.JSONUtil;
import com.audaque.springboot.foshanupload.web.amqpdemo.model.rsp.SysUserMpgtRsp;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @author zgb
 * @desc ...
 * @date 2024-01-07 05:24:14
 */

@Slf4j
@RestController
@RequestMapping("/consumerTemplate")
public class ConsumerTemplateController {
  @Resource
  private RabbitTemplate rabbitTemplate;
  @Autowired
  private AmqpTemplate amqpTemplate;


  @PostMapping("/consumerPullDiy")
  public void consumerPullDiy() {
    // 要拉取的消息条数
    int batchSize = 2;
    rabbitTemplate.execute(
    channel -> {
      List<SysUserMpgtRsp> messageBodyList = new ArrayList<>(batchSize);
      long deliveryTag = 0;

      while (messageBodyList.size() < batchSize) {
        GetResponse response = channel.basicGet("sun.pull.queue", false);
        log.debug("The result of pulling the message response this time: {}", response);
        if (Objects.isNull(response)) {
          break;
        }

        byte[] bodyBytes = response.getBody();
        String json = new String(bodyBytes);
        SysUserMpgtRsp messageBody = JSONUtil.toBean(json, SysUserMpgtRsp.class);
        messageBodyList.add(messageBody);
        deliveryTag = response.getEnvelope().getDeliveryTag();
      }

      log.debug("A total of {} messages were pulled this time：{}", messageBodyList.size(), messageBodyList);

      try {
        // 可以在此做一些业务操作
        // 也可以将此消费者测试用例封装起来，用java.util.function.Consumer作为参数传入，在此进行consumer.accept(messageBodyList)，以确保业务操作成功之后ack
        //处理业务
        //处理业务后应答
        channel.basicAck(deliveryTag, true);
      } catch (Exception e) {
        log.error("Abnormal consumption!", e);

        try {
          channel.basicNack(deliveryTag, true, true);
        } catch (Exception ex) {
          log.error("Reject message exception!", ex);
        }
      }

      // Or you can return the result to handle it yourself,
      // but you must ensure that the ack will not affect you
      return null;
    });
  }

  @PostMapping("/consumerPull")
  public void consumerPull() {
    Object o = rabbitTemplate.receiveAndConvert("sun.pull.queue");
    String jsonStr = JSONUtil.toJsonStr(o);
    SysUserMpgtRsp bean = JSONUtil.toBean(jsonStr, SysUserMpgtRsp.class);
    System.out.println(bean);
  }


}
